package com.deep.flink.demo.job._1_binlog;

import lombok.SneakyThrows;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *
 */
public class _2_MySqlBinlogJob {

    @SneakyThrows
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //========== 【表环境】 ==========
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        //========== 【注册表】 ==========
        String source = createSourceDDL(tableEnv);
        String sink = createSinkDDL(tableEnv);
        //========== 【】 ==========
        StringBuffer sql = new StringBuffer();
        sql.append("insert into ").append(sink)
                .append(" select * from ").append(source);
        StatementSet st = tableEnv.createStatementSet();
        st.addInsertSql(sql.toString());
        //========== 【】 ==========
        st.execute();
    }

    public static String createSourceDDL(StreamTableEnvironment tableEnv) {
        //语句
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE user_base (")
                .append("ub_id          BIGINT, ")
                .append("ub_login_name  STRING, ")
                .append("ub_login_pwd   STRING, ")
                .append("ub_begin_time  TIMESTAMP, ")
                .append("PRIMARY KEY (ub_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'mysql-cdc', ")
                .append("'hostname' = '127.0.0.1', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'port' = '3306', ")
                .append("'database-name' = 'ds_canal', ")
                .append("'table-name'= 'user_base', ")
                .append("'server-id' = '333', ")
                .append("'server-time-zone' = 'Asia/Shanghai', ")
                .append("'debezium.database.whitelist' = 'ds_canal', ")
                .append("'debezium.table.whitelist' = 'ds_canal.user_base', ")
                .append("'debezium.include.schema.changes' = 'false' ")
                .append(")");
        //注册表
        TableResult tblResult = tableEnv.executeSql(ddl.toString());

        return "user_base";
    }

    public static String createSinkDDL(StreamTableEnvironment tableEnv) {
        //语句
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE user_base_bak (")
                .append("ub_id          BIGINT, ")
                .append("ub_login_name  STRING, ")
                .append("ub_login_pwd   STRING, ")
                .append("ub_begin_time  TIMESTAMP, ")
                .append("PRIMARY KEY (ub_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'jdbc', ")
                .append("'url' = 'jdbc:mysql://127.0.0.1:3306/ds_canal?serverTimezone=UTC', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'table-name' = 'user_base_bak', ")
                .append("'driver' = 'com.mysql.jdbc.Driver' ")
                // .append("'sink.buffer-flush.max-rows'= ''")
                //.append("'sink.buffer-flush.interval'= ''")
                .append(")");
        //注册表
        TableResult tblResult = tableEnv.executeSql(ddl.toString());

        return "user_base_bak";
    }
}
